package com.erbao.io.netty;

import com.erbao.IoUtil;
import com.erbao.SleepUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import lombok.Data;
import org.junit.Test;

import java.io.*;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 1.需求：写一个RPC:
 * 2.来回通信，连接数量，拆包、解包
 * 3.动态代理，序列化，反序列化，自定义协议封装
 * 4.客户端连接池
 * 5.RPC通俗解释：像调用本地方法一样调用远程方法，Java的面向接口开发，采用动态代理（feign是不是这么实现的？）
 * @author CaoBaoYin
 * @since 22/10/22 19:27
 */
public class MyRPCTest {

    @Test
    public void startServer(){
        MyCar myCar = new MyCar();
        MyFly myFly = new MyFly();
        Dispatcher.register(Car.class.getName(), myCar);
        Dispatcher.register(Fly.class.getName(), myFly);

        NioEventLoopGroup boss = new NioEventLoopGroup(50);
        NioEventLoopGroup worker = boss;

        ServerBootstrap server = new ServerBootstrap();
        ChannelFuture bind = server.group(boss, worker).channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        System.out.println("server accept port:"+channel.remoteAddress().getPort());
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new PkgMsgDecoder());
                        pipeline.addLast(new ServerRequestHandler());
                    }
                }).bind(new InetSocketAddress("localhost", 9090));

        try {
            // 第一个sync确保绑定成功，第二个sync服务端一直等待关闭，保证服务阻塞
            bind.sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void get () {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        for (int i = 0; i < 20; i++) {
            new Thread(()->{
                Car car = proxyGet(Car.class);
                String arg = "hahha-" + atomicInteger.incrementAndGet();
                String result = car.xxoo(arg);
                System.out.println(result+" client发送参数："+arg);
            }).start();
        }

        // System.out.println("执行结束！！！");

        SleepUtil.sleep(3000);

    }

    private <T> T proxyGet(Class<T> interfaceClazz) {
        Object instance = Proxy.newProxyInstance(interfaceClazz.getClassLoader(), new Class[]{interfaceClazz}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 接口中的任何方法调用都会调用invoke
                System.out.println("进入代理");
                System.out.println("proxy: "+proxy.getClass().getName()+" method: "+method.getName()+"args:");
                // toString 会造成递归调用
                // System.out.println("proxy:".concat(proxy.toString()).concat(" method:").concat(method.toString()).concat(" args:").concat(args.toString()));
                // method.invoke(proxy, args) 这么写还是会调用当前方法

                // 设计consumer调用provider的过程
                // 1.调用服务，方法，参数 ==》 封装message (content 即body)
                String interfaceName = interfaceClazz.getName();
                String methodName = method.getName();
                Class<?>[] parameterTypes = method.getParameterTypes();

                MyContent content = new MyContent();
                content.setName(interfaceName);
                content.setMethodName(methodName);
                content.setArgs(args);
                content.setParameterTypes(parameterTypes);

                // 对象转成字节数组，channle 发送使用 bytebuf。对象进，出字节数组
                /* todo 此处没有关闭流 */
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                ObjectOutputStream oos = new ObjectOutputStream(baos);//分析此处的嵌套，装饰模式（里面的对象先执行，外面的对象先调用）
                oos.writeObject(content);
                byte[] msgBodyBytes = baos.toByteArray();

                /*2.requestId + message ，本地要缓存requestId
                * 自定义协议：【header：请求相关的信息】+【body：msgBody】*/
                MyHeader header = createHeader(msgBodyBytes);
                baos.reset();
                // oos.reset(); //msgHeaderBytes.length = 100
                oos = new ObjectOutputStream(baos); //msgHeaderBytes.length = 103
                oos.writeObject(header);
                byte[] msgHeaderBytes = baos.toByteArray();

                // 3.连接池获取连接
                InetSocketAddress address = new InetSocketAddress("127.0.0.1", 9090);
                NioSocketChannel client = ClientFactory.INSTANCE.getClient(address);

                /*注册响应，即响应后代码继续执行*/
                CompletableFuture completableFuture = new CompletableFuture<String>();
                ResponseMappingCallBack.addCallback(header.getRequestId(), completableFuture);

                // 4.发送 走的IO流 netty（event 驱动）
                System.out.println("发送的header和body的length=="+msgHeaderBytes.length + ":" +msgBodyBytes.length);
                ByteBuf reqBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeaderBytes.length + msgBodyBytes.length);
                reqBuf.writeBytes(msgHeaderBytes);
                reqBuf.writeBytes(msgBodyBytes);
                client.writeAndFlush(reqBuf).sync(); //流肯定是双向的，这行代码结束只代表out结束，还需要等待响应，就需要阻塞住

                // 5.走IO流 响应了对应同一个请求，代码回到对应位置再继续执行
                return completableFuture.get();
            }
        });
        return (T) instance;
    }

    private MyHeader createHeader(byte[] msgBodyBytes) {
        MyHeader header = new MyHeader();
        header.setContentLength(msgBodyBytes.length);
        header.setFlag(0x141a1f14);//十六进制，一个数占四位
        header.setRequestId(UUID.randomUUID().getLeastSignificantBits());

        return header;
    }

}

enum ClientFactory{
    INSTANCE;
    int poolSize = 1;
    NioEventLoopGroup clientWorker;

    Random random = new Random();
    /*一个consumer（client）可以连接很多的provider（server），每一个server都有自己的pool
    * key server的ip+port value: */
    ConcurrentHashMap<InetSocketAddress, ClientPool> oubBoxs = new ConcurrentHashMap<>();

    public NioSocketChannel getClient (InetSocketAddress address) {
        ClientPool clientPool = oubBoxs.get(address);
        // System.out.println("首次获取连接池[key:"+address+"]结果："+clientPool);
        /*获取对象判空，要考虑并发情况，双重检查
        * 但此处这种写法类似双重检查：1.外层if；2.if absent*/
        if (clientPool == null) {
            oubBoxs.putIfAbsent(address, new ClientPool(poolSize));
        }

        /*取出有效client*/
        clientPool = oubBoxs.get(address);
        // System.out.println("再次获取连接池[key:"+address+"]结果："+clientPool);
        int index = random.nextInt(poolSize);
        NioSocketChannel client = clientPool.clients[index];
        // System.out.println("首次获取连接池中位置【"+index+"】的连接："+client);
        if (client!=null && client.isActive()) {
            return client;
        } else {
            /*对当前client进行初始化 首次进入肯定是因为null*/
            synchronized (clientPool.locks[index]) {
                // 此处需要double check 否则连接池不管用
                client = clientPool.clients[index];
                // System.out.println("再次获取连接池中位置【"+index+"】的连接："+client);
                if (client!=null && client.isActive()) {
                    return client;
                }

                // System.out.println("创建连接放入位置【"+index+"】并返回");
                return client = clientPool.clients[index] = createClient (address);
            }
        }

        // 获取到了not active的client
        /*if (client!=null && !client.isActive()) {
            return getClient(address);
        }*/



    }

    private NioSocketChannel createClient (InetSocketAddress address) {
        // 基于netty的客户端
        clientWorker = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(clientWorker).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new PkgMsgDecoder());
                pipeline.addLast(new ClientResponseHandler());
            }
        }).connect(address);

        try {
            NioSocketChannel channel = (NioSocketChannel) connect.sync().channel();
            return channel;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

class ResponseMappingCallBack {
    /*key requestId,*/
    static ConcurrentHashMap<Long, CompletableFuture> mapping = new ConcurrentHashMap<>();

    public static void addCallback(long requestId, CompletableFuture completableFuture) {
        mapping.put(requestId, completableFuture);
    }

    public static void runCallback(MyPkgMsg pkgMsg){
        long requestId = pkgMsg.header.requestId;
        mapping.get(requestId).complete(pkgMsg.content.responseStr);
        removeCallback(requestId);
    }

    public static void removeCallback(long requestId){
        mapping.remove(requestId);
    }
}

class PkgMsgDecoder extends ByteToMessageDecoder {

    // 父类里一定有channelread { 会有byte buf处理的环绕，decode之后剩余留存 } > byte buf
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        // 请求的数据，服务端读取
        // System.out.println(" 读取之前，发现可读字节数【"+byteBuf.readableBytes()+"】");
        while (byteBuf.readableBytes() >= MyHeader.BYTES_LENGTH) {// 消息头
            byte[] bytes = new byte[MyHeader.BYTES_LENGTH];
            // byteBuf.readBytes(bytes);// 这种读取方式会移动读指针
            byteBuf.getBytes(byteBuf.readerIndex(), bytes);// 获取数据，不移动指针
            ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
            ObjectInputStream ois = new ObjectInputStream(bais);
            MyHeader header = (MyHeader) ois.readObject();

            // System.out.println(" 读取的请求header：" + header);

            /*读取消息体*/
            if (byteBuf.readableBytes() >= header.getContentLength()) {
                // 进入判断，可以获取一个完整请求的数据，需要移动指针
                byteBuf.readBytes(MyHeader.BYTES_LENGTH);

                byte[] contentBytes = new byte[(int) header.getContentLength()];
                byteBuf.readBytes(contentBytes);
                ByteArrayInputStream din = new ByteArrayInputStream(contentBytes);
                ObjectInputStream doin = new ObjectInputStream(din);
                MyContent content = (MyContent) doin.readObject();

                if (header.flag == 0x14141414) {
                    System.out.println("请求数据，server解码中……");
                } else if (header.flag == 0x14141424) {
                    System.out.println("响应数据，client解码中……");
                }

                list.add(new MyPkgMsg(header, content));
                // System.out.println(" 读取的请求内容：" + content);
            } else {
                // System.out.println(" 发现剩余可读大小【" + byteBuf.readableBytes() + "】小于header中显示的content的大小【" + header.getContentLength() + "】");
                break;// 剩余的数据会自动拼接缓存，也就是自动暂存
            }
        }

    }
}

/*服务端*/
class ServerRequestHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof MyPkgMsg) {
            System.out.println("读取到msg："+msg);
            MyPkgMsg requestMsg = (MyPkgMsg) msg;

            String ioThreadName = Thread.currentThread().getName();
            /*io线程和业务线程
            * 1.直接在当前线程处理io和业务
            * 2.使用netty自己的eventloop来处理业务及返回
            * 3.也可以使用自己的线程池。*/

            // 把现有的任务使用已有的线程进行执行。因为一台机器的cpu有限，能开启的线程数也是有限的。
            // ctx.executor().execute(()->{// io线程和业务线程是一个。因为一个线程处理一个连接的所有事件,一个线程一个eventloop，不用加锁.一个selector绑定一个线程处理一个连接线性读写
            ctx.executor().parent().next().execute(()->{// executor eventloop,parent eventloopgroup。从group中挑选下一个线程执行
                String execTreadName = Thread.currentThread().getName();
                MyContent content = new MyContent();
                // 查看io线程和业务线程是否一样。执行的参数和调用的参数，要看是不是一个参数，判断通信有没有错位
                String responseStr = "IO线程：".concat(ioThreadName).concat(" 执行线程：").concat(execTreadName).concat(" server接收参数：").concat(Objects.toString(requestMsg.content.args[0], ""));
                System.out.println(responseStr);

                MyContent requestContent = requestMsg.getContent();
                Object object = Dispatcher.get(requestContent.getName());
                Object result = null;
                try {
                    Method method = object.getClass().getMethod(requestContent.getMethodName(), requestContent.getParameterTypes());
                    result = method.invoke(object, requestContent.getArgs());
                } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
                    throw new RuntimeException(e);
                }

                content.setResponseStr(result);
                byte[] bodyBytes = IoUtil.serialize(content);

                MyHeader header = new MyHeader();
                header.setRequestId(requestMsg.header.requestId);
                header.setFlag(0x14141424);
                header.setContentLength(bodyBytes.length);
                byte[] headerBytes = IoUtil.serialize(header);
                System.out.println("header length:"+headerBytes.length+"body length:"+bodyBytes.length);

                ByteBuf responseByteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(headerBytes.length + bodyBytes.length);
                responseByteBuf.writeBytes(headerBytes);
                responseByteBuf.writeBytes(bodyBytes);

                ctx.writeAndFlush(responseByteBuf);
            });


        }


    }
}
class ClientResponseHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 响应回来的，要读的。
        if (msg instanceof MyPkgMsg) {
            MyPkgMsg responseMsg = (MyPkgMsg) msg;
            ResponseMappingCallBack.runCallback(responseMsg);
        }

        super.channelRead(ctx, msg);
    }
}

class ClientPool {
    NioSocketChannel[] clients;
    Object[] locks;

    /**
     *
     * @param size 线程池大小
     */
    public ClientPool(int size) {
        clients = new NioSocketChannel[size];
        locks = new Object[size];

        for (int i = 0; i < size; i++) {
            locks[i] = new Object();

        }
    }
}

@Data
class MyHeader implements Serializable {
    public static final int BYTES_LENGTH = 103; // header 序列化后的长度跟jdk版本、类全名（包的长度）也有很关系
    // 值 标识什么样的协议 用二进制位标识 int 四个字节 32位可以表示很多信息
    int flag;//4
    long requestId;//8
    long contentLength;//8
}

@Data
class MyContent implements Serializable {
    String name;
    String methodName;
    Class<?>[] parameterTypes;
    Object[] args;
    Object responseStr;
}

@Data
class MyPkgMsg {
    MyHeader header;
    MyContent content;

    public MyPkgMsg(MyHeader header, MyContent content) {
        this.header = header;
        this.content = content;
    }
}

class Dispatcher {
    public static ConcurrentHashMap<String, Object> invokeMap = new ConcurrentHashMap<>();

    public static void register(String key, Object value) {
        invokeMap.put(key, value);
    }

    public static Object get (String key) {
        return invokeMap.get(key);
    }
}

class MyCar implements Car{

    @Override
    public String xxoo(String msg) {
        System.out.println("car 接受到了：".concat(msg));
        return "car 已经处理了：".concat(msg);
    }
}

class MyFly implements Fly{

    @Override
    public void ooxx(String msg) {
        System.out.println("fly 接收到了：".concat(msg));
    }
}

interface Car {
    String xxoo(String msg);
}

interface Fly {
    void ooxx(String msg);
}